package com.zlsy.springbootdemo.redis.consumer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.Map;

/**
 * @author zhouliang
 * @date 2020/12/15 17:38
 */
@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class StreamMessageListener
    implements StreamListener<String, MapRecord<String, String, String>> {

  private final RedisTemplate<String, Serializable> redisTemplate;

  @Override
  public void onMessage(MapRecord<String, String, String> message) {
    String stream = message.getStream();
    System.out.println(stream);
    RecordId recordId = message.getId();
    Map<String, String> map = message.getValue();
    Long sequence = recordId.getSequence();
    Long timestamp = recordId.getTimestamp();
    String value = recordId.getValue();
    log.info("map:{},sequence:{},timestamp:{},value:{}", map, sequence, timestamp, value);

    Long myStreams = redisTemplate.opsForStream().acknowledge("myStreams", message);
    log.info("acknowledge:{}", myStreams);
  }
}
